package com.zyx.kafkademo.admin;

import com.zyx.kafkademo.common.CommonConfig;
import org.apache.kafka.clients.admin.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author Yaxi.Zhang
 * @since 2021/7/22 10:48
 * desc: KafkaAdmin操作主题案例
 */
public class KafkaAdminTopicOperationTest {

    private static AdminClient client;
    private static final String KAFKA_TOPIC = "kafka_topic_wc";

    @Before
    public void init() {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CommonConfig.COMPANY_KAFKA_SERVER);
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        client = AdminClient.create(props);
    }

    @After
    public void stop() {
        client.close();
    }


    @Test
    public void describeTopic() {


        DescribeTopicsResult result = client.describeTopics(Collections.singleton(KAFKA_TOPIC));
        try {
            Map<String, TopicDescription> descriptionMap = result.all().get();
            System.out.println(descriptionMap.get(KAFKA_TOPIC));
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        client.close();
    }


    @Test
    public void listTopics() {
        try {
            client.listTopics()
                    .listings()
                    .get(1, TimeUnit.MINUTES)
                    .forEach(topicListing -> {
                        System.out.println("currentTopic is: " + topicListing.name());
                    });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTopic() {
        // 参数: 主题名称, 分区数目, 副本数目
        NewTopic newTopic = new NewTopic(KAFKA_TOPIC, 1, (short) 2);

        CreateTopicsResult result = client.createTopics(Collections.singleton(newTopic));
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTopicWithAssignments() {
        // 指定分区副本分配策略, 这里创建了一个分区数为4, 副本数为1的分区副本分配策略
        Map<Integer, List<Integer>> replicasAssignments = new HashMap<>(1 << 4);
        replicasAssignments.put(0, Collections.singletonList(0));
        replicasAssignments.put(1, Collections.singletonList(0));
        replicasAssignments.put(2, Collections.singletonList(0));
        replicasAssignments.put(3, Collections.singletonList(0));

        NewTopic newTopic = new NewTopic(KAFKA_TOPIC, replicasAssignments);

        CreateTopicsResult result = client.createTopics(Collections.singleton(newTopic));
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTopicWithConfig() {
        String topic = "topic-admin";

        NewTopic newTopic = new NewTopic(topic, 2, (short) 1);

        Map<String, String> configs = new HashMap<>(1 << 4);
        // 指定压缩策略, 默认为delete, 可以修改为compact
        configs.put("cleanup.policy", "compact");
        newTopic.configs(configs);

        CreateTopicsResult result = client.createTopics(Collections.singleton(newTopic));
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void deleteTopic() {
        String topic = "topic-admin";
        try {
            client.deleteTopics(Collections.singleton(topic)).all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }
}
